package com.qianbao.finance.sync.modules.elasticsearch.service.impl;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.qianbao.finance.sync.modules.elasticsearch.model.DatabaseTableModel;
import com.qianbao.finance.sync.modules.elasticsearch.model.IndexTypeModel;
import com.qianbao.finance.sync.modules.elasticsearch.model.SyncMappingConfig;
import com.qianbao.finance.sync.modules.elasticsearch.model.request.InitMappingRequest;
import com.qianbao.finance.sync.modules.elasticsearch.service.BaseDaoService;
import com.qianbao.finance.sync.modules.elasticsearch.service.ElasticsearchService;
import com.qianbao.finance.sync.modules.elasticsearch.service.MappingService;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.PropertySource;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.sql.Timestamp;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Optional;
import java.util.concurrent.*;
import java.util.stream.Collectors;


@Service
@PropertySource("classpath:/config/mapping.properties")
@ConfigurationProperties
public class MappingServiceImpl implements MappingService, InitializingBean {
    private static final Logger logger = LoggerFactory.getLogger(MappingServiceImpl.class);
    private static final DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    private Map<String, String> dbEsMapping;
    private Map<DatabaseTableModel, IndexTypeModel> dbEsBiMapping;
    private Map<String, String> tablePrimaryKeyMap;
    private Map<String, String> dbEsDataTypeMapping;
    private Map<String, Converter> mysqlTypeElasticsearchTypeMapping;
    private List<SyncMappingConfig> syncMappingConfigList;
    private Map<String, SyncMappingConfig> syncMappingConfigMap;
    private ExecutorService cachedThreadPool;
    @Resource
    private BaseDaoService baseDaoService;
    @Resource
    private ElasticsearchService elasticsearchService;
    @Override
    public Map<String, String> getTablePrimaryKeyMap() {
        return tablePrimaryKeyMap;
    }
    @Override
    public void setTablePrimaryKeyMap(Map<String, String> tablePrimaryKeyMap) {
        this.tablePrimaryKeyMap = tablePrimaryKeyMap;
    }

    @Override
    public IndexTypeModel getIndexType(DatabaseTableModel databaseTableModel) {
        return dbEsBiMapping.get(databaseTableModel);
    }

    @Override
    public DatabaseTableModel getDatabaseTableModel(IndexTypeModel indexTypeModel) {
       // return dbEsBiMapping.inverse().get(indexTypeModel);
        return null;
    }

    @Override
    public Object getElasticsearchTypeObject(String mysqlType, String data) {
        Optional<Entry<String, Converter>> result = mysqlTypeElasticsearchTypeMapping.entrySet().parallelStream().filter(entry -> mysqlType.toLowerCase().contains(entry.getKey())).findFirst();
        return (result.isPresent() ? result.get().getValue() : (Converter) data1 -> data1).convert(data);
    }

    @Override
    public String initMapping(InitMappingRequest request) {
        dbEsMapping.forEach((key, value) -> {
            String[] keyStrings = StringUtils.split(key, ".");
            String[] valueStrings = StringUtils.split(value, ".");
            //dbEsBiMapping.put(new DatabaseTableModel(keyStrings[0], keyStrings[1]), new IndexTypeModel(valueStrings[0], valueStrings[1]));
            cachedThreadPool.submit(() -> {
                List<Map<String, Object>> columnList = baseDaoService.selectColumnsBySchemaAndTable(keyStrings[0], keyStrings[1]);
                if (!CollectionUtils.isEmpty(columnList)) {
                    Map<String, Map<String, Map<String, String>>> propsMap = Maps.newHashMap();
                    Map<String, Map<String, String>> columnsMap = Maps.newHashMap();
                    propsMap.put("properties", columnsMap);
                    columnList.forEach(item -> {
                        Map<String, String> columnPropsMap = Maps.newHashMap();
                        columnsMap.put(item.get("COLUMN_NAME").toString(), columnPropsMap);
                        columnPropsMap.put("type", dbEsDataTypeMapping.get(item.get("DATA_TYPE")));
                    });
                    elasticsearchService.createMapping(valueStrings[0], valueStrings[1], propsMap);
                }
            });
        });
/*        syncMappingConfigList.stream().parallel().forEach(syncMappingConfig -> {
            List<Map<String, Object>> columnList = baseDao.selectColumnsBySchemaAndTable(syncMappingConfig.getDatabaseName(), syncMappingConfig.getTableName());
            if(!CollectionUtils.isEmpty(columnList)){
                Map<String,Map<String,Map<String,String>>> propsMap = Maps.newHashMap();
                Map<String,Map<String,String>> columnsMap = Maps.newHashMap();
                propsMap.put("properties", columnsMap);
                columnList.forEach(item->{
                    Map<String,String> columnPropsMap = Maps.newHashMap();
                    columnsMap.put(item.get("COLUMN_NAME").toString(),columnPropsMap);
                    columnPropsMap.put("type", dbEsDataTypeMapping.get(item.get("DATA_TYPE")));
                });
                elasticsearchService.createMapping(syncMappingConfig.getEsIndex(), syncMappingConfig.getEsType(), propsMap);
            }
        });*/
        return "在ES中初始化mapping信息完成，请自行核对mapping信息是否符合预期";
    }

    @Override
    public String syncByConfig(Integer stepSize) {
        dbEsMapping.forEach((key, value) -> {
            String[] keyStrings = StringUtils.split(key, ".");
            logger.info("开始同步库：[{}],表：[{}]", keyStrings[0], keyStrings[1]);
            IndexTypeModel indexTypeModel = getIndexType(new DatabaseTableModel(keyStrings[0], keyStrings[1]));
            String primaryKey = Optional.ofNullable(getTablePrimaryKeyMap().get(keyStrings[0] + "." + keyStrings[1])).orElse("id");
            if(!primaryKey.equals("id") && !primaryKey.equals("order_id") && !primaryKey.equals("account_id")){
                logger.info(keyStrings[0] + "." + keyStrings[1]+"主键特殊，忽略导入");
            }else if (indexTypeModel == null) {
                throw new IllegalArgumentException(String.format("配置文件中缺失database=%s和table=%s所对应的index和type的映射配置", keyStrings[0], keyStrings[1]));
            }else {
                cachedThreadPool.submit(() -> {
                    try {
                        long maxPK = baseDaoService.selectMaxPK(primaryKey, keyStrings[0], keyStrings[1]);
                        long minPK = baseDaoService.selectMinPK(primaryKey, keyStrings[0], keyStrings[1]);
                        for (long i = minPK; i < maxPK+1; ) {
                            i = batchInsertElasticsearch(keyStrings[0], keyStrings[1], primaryKey, i, stepSize, indexTypeModel);
                            logger.info(String.format("当前同步database=%s,table=%s, pk=%s，总共total=%s，进度=%s%%", keyStrings[0], keyStrings[1], i, maxPK, new BigDecimal(i * 100).divide(new BigDecimal(maxPK), 3, BigDecimal.ROUND_HALF_UP)));
                            updateSyncMappingConfig(keyStrings[0], keyStrings[1], stepSize);
                        }
                        logger.info(String.format("同步database=%s,table=%s完成", keyStrings[0], keyStrings[1]));
                    } catch (Exception e) {
                        logger.error("批量转换并插入Elasticsearch异常,database:[{}],table:[{}]", keyStrings[0], keyStrings[1], e);
                    }
                });
            }
        });
        return "已经开始同步数据，请观察页面数据变化情况";
    }

    @Override
    public String removeAll() {
        dbEsMapping.forEach((key, value) -> {
            String[] keyStrings = StringUtils.split(key, ".");
            String[] valueStrings = StringUtils.split(value, ".");
            cachedThreadPool.submit(() -> {
                elasticsearchService.deleteIndex(valueStrings[0]);
                updateSyncMappingConfig(keyStrings[0], keyStrings[1], 1);
            });
        });
        return "清空配置相关的es中index完成";
    }


    @Override
    public String removeEsByDataBase(String databaseName) {
        dbEsMapping.forEach((key, value) -> {
            String[] keyStrings = StringUtils.split(key, ".");
            String[] valueStrings = StringUtils.split(value, ".");
            String index = keyStrings[0];
            if(index.equals(databaseName)){
                cachedThreadPool.submit(() -> {
                    elasticsearchService.deleteIndex(valueStrings[0]);
                    updateSyncMappingConfig(keyStrings[0], keyStrings[1], 1);
                });
            }
        });
        return "清空配置相关的es中index完成";
    }


    @Override
    public String initMappingByDataBase(String databaseName) {
        dbEsMapping.forEach((key, value) -> {
            String[] keyStrings = StringUtils.split(key, ".");
            String[] valueStrings = StringUtils.split(value, ".");
            String index = keyStrings[0];
            if(index.equals(databaseName)) {
                cachedThreadPool.submit(() -> {
                    List<Map<String, Object>> columnList = baseDaoService.selectColumnsBySchemaAndTable(keyStrings[0], keyStrings[1]);
                    if (!CollectionUtils.isEmpty(columnList)) {
                        Map<String, Map<String, Map<String, String>>> propsMap = Maps.newHashMap();
                        Map<String, Map<String, String>> columnsMap = Maps.newHashMap();
                        propsMap.put("properties", columnsMap);
                        columnList.forEach(item -> {
                            Map<String, String> columnPropsMap = Maps.newHashMap();
                            columnsMap.put(item.get("COLUMN_NAME").toString(), columnPropsMap);
                            columnPropsMap.put("type", dbEsDataTypeMapping.get(item.get("DATA_TYPE")));
                        });
                        elasticsearchService.createMapping(valueStrings[0], valueStrings[1], propsMap);
                    }
                });
            }
        });
        return "在ES中初始化mapping信息完成，请自行核对mapping信息是否符合预期";
    }

    @Override
    public String byConfigByDataBase(String databaseName) {
        dbEsMapping.forEach((key, value) -> {
            String[] keyStrings = StringUtils.split(key, ".");
            logger.info("开始同步库：[{}],表：[{}]", keyStrings[0], keyStrings[1]);
            IndexTypeModel indexTypeModel = getIndexType(new DatabaseTableModel(keyStrings[0], keyStrings[1]));
            String primaryKey = Optional.ofNullable(getTablePrimaryKeyMap().get(keyStrings[0] + "." + keyStrings[1])).orElse("id");
            if(!primaryKey.equals("id") && !primaryKey.equals("order_id") && !primaryKey.equals("account_id")){
                logger.info(keyStrings[0] + "." + keyStrings[1]+"主键特殊，忽略导入");
            }else if (indexTypeModel == null) {
                throw new IllegalArgumentException(String.format("配置文件中缺失database=%s和table=%s所对应的index和type的映射配置", keyStrings[0], keyStrings[1]));
            }else {
                String index = keyStrings[0];
                if(index.equals(databaseName)) {
                    cachedThreadPool.submit(() -> {
                        try {
                            long maxPK = baseDaoService.selectMaxPK(primaryKey, keyStrings[0], keyStrings[1]);
                            long minPK = baseDaoService.selectMinPK(primaryKey, keyStrings[0], keyStrings[1]);
                            for (long i = minPK; i < maxPK + 1; ) {
                                i = batchInsertElasticsearch(keyStrings[0], keyStrings[1], primaryKey, i, 500, indexTypeModel);
                                logger.info(String.format("当前同步database=%s,table=%s, pk=%s，总共total=%s，进度=%s%%", keyStrings[0], keyStrings[1], i, maxPK, new BigDecimal(
                                        i * 100).divide(new BigDecimal(maxPK), 3, BigDecimal.ROUND_HALF_UP)));
                                updateSyncMappingConfig(keyStrings[0], keyStrings[1], 500);
                            }
                            logger.info(String.format("同步database=%s,table=%s完成", keyStrings[0], keyStrings[1]));
                        } catch (Exception e) {
                            logger.error("批量转换并插入Elasticsearch异常,database:[{}],table:[{}]", keyStrings[0], keyStrings[1], e);
                        }
                    });
                }
            }
        });
        return "已经开始同步数据，请观察页面数据变化情况";
    }





    @Transactional(readOnly = true, rollbackFor = Exception.class)
    long batchInsertElasticsearch(String databaseName, String tableName, String primaryKey, long from, long stepSize, IndexTypeModel indexTypeModel) {
        List<Map<String, Object>> dataList = baseDaoService.selectByPKIntervalLockInShareModeStep(primaryKey, from, stepSize, databaseName, tableName);
        dataList = convertDateType(dataList, primaryKey);
        long maxIdValue = Long.parseLong(dataList.remove(dataList.size()-1).get("maxId").toString());
        Map<String, Map<String, Object>> dataMap = dataList.parallelStream().collect(Collectors.toMap(strObjMap -> String.valueOf(strObjMap.get(primaryKey)), map -> map));
        elasticsearchService.batchInsertById(indexTypeModel.getIndex(), indexTypeModel.getType(), dataMap);
        return maxIdValue+1;
    }


    private List<Map<String, Object>> convertDateType(List<Map<String, Object>> source, String primaryKey) {
        Map<String, Object> maxIdMap = Maps.newHashMap();
        maxIdMap.put("maxId", -10000L);
        source.parallelStream().forEach(map -> map.forEach((key, value) -> {
            if (value instanceof Timestamp) {
                map.put(key, LocalDateTime.ofInstant(((Timestamp) value).toInstant(), ZoneId.systemDefault()));
            }
            if (key.equals(primaryKey)){
                if(Long.parseLong(value.toString()) > Long.parseLong(maxIdMap.get("maxId").toString())){
                    maxIdMap.put("maxId", Long.parseLong(value.toString()));
                }
            }
        }));
        source.add(maxIdMap);
        return source;
    }

    @Override
    public boolean makeMappingStep(String database) {
        List<Map<String, Object>> columnList = baseDaoService.selectTableName(database);
        columnList.forEach(mapEntry-> System.out.println("dbEsMapping["+database+"."+mapEntry.get("TABLE_NAME").toString()+"]="+StringUtils.remove(mapEntry.get("TABLE_NAME").toString(), "_")+"."+StringUtils.remove(mapEntry.get("TABLE_NAME").toString(), "_")));
        columnList.forEach(mapEntry-> System.out.println("tablePrimaryKeyMap["+database+"."+mapEntry.get("TABLE_NAME").toString()+"]=id"));
        return true;
    }

    @Override
    public boolean makeGoMappingStep(String database) {
        List<Map<String, Object>> columnList = baseDaoService.selectTableName(database);
        StringBuilder sbTables = new StringBuilder("tables = [");
        StringBuilder sbIndexs = new StringBuilder();
        columnList.forEach(mapEntry->{
            sbTables.append("\"").append(mapEntry.get("TABLE_NAME").toString()).append("\", ");
            sbIndexs.append("schema = \"").append(database).append("\"\r\n")
                    .append("table = \"").append(mapEntry.get("TABLE_NAME").toString()).append("\"\r\n")
                    .append("index = \"").append("go"+StringUtils.remove(mapEntry.get("TABLE_NAME").toString(), "_")).append("\"\r\n")
                    .append("type= \"").append(StringUtils.remove(mapEntry.get("TABLE_NAME").toString(), "_")).append("\"\r\n");
        });
        sbTables.delete(sbTables.length()-2, sbTables.length()).append("]");
        System.out.println(sbTables.toString());
        System.out.println(sbIndexs.toString());
        return true;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        dbEsBiMapping = new HashMap<>();
        syncMappingConfigList = Lists.newArrayList();
        syncMappingConfigMap = Maps.newHashMap();
        dbEsMapping.forEach((key, value) -> {
            String[] keyStrings = StringUtils.split(key, ".");
            String[] valueStrings = StringUtils.split(value, ".");
            dbEsBiMapping.put(new DatabaseTableModel(keyStrings[0], keyStrings[1]), new IndexTypeModel(valueStrings[0], valueStrings[1]));
            SyncMappingConfig configEntry = new SyncMappingConfig();
            configEntry.setDatabaseName(keyStrings[0]);
            configEntry.setTableName(keyStrings[1]);
            configEntry.setEsIndex(valueStrings[0]);
            configEntry.setEsType(valueStrings[1]);
            syncMappingConfigList.add(configEntry);
            syncMappingConfigMap.put(key, configEntry);
        });

        mysqlTypeElasticsearchTypeMapping = Maps.newHashMap();
        mysqlTypeElasticsearchTypeMapping.put("char", data -> data);
        mysqlTypeElasticsearchTypeMapping.put("text", data -> data);
        mysqlTypeElasticsearchTypeMapping.put("blob", data -> data);
        mysqlTypeElasticsearchTypeMapping.put("int", Long::valueOf);
        mysqlTypeElasticsearchTypeMapping.put("date", data -> LocalDateTime.parse(data, formatter));
        mysqlTypeElasticsearchTypeMapping.put("time", data -> LocalDateTime.parse(data, formatter));
        mysqlTypeElasticsearchTypeMapping.put("float", Double::valueOf);
        mysqlTypeElasticsearchTypeMapping.put("double", Double::valueOf);
        mysqlTypeElasticsearchTypeMapping.put("decimal", Double::valueOf);

        cachedThreadPool = new ThreadPoolExecutor(10, 30, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), (ThreadFactory) Thread::new);
    }

    public Map<String, String> getDbEsMapping() {
        return dbEsMapping;
    }

    @Override
    public List<SyncMappingConfig> listDbEsMapping() {
        logger.debug("开始刷新【{}】", System.currentTimeMillis());
        syncMappingConfigList.parallelStream().forEach(configEntry->{
            //if(configEntry.getChangedCount() > 0) {
                configEntry.setMysqlCount(baseDaoService.count(configEntry.getDatabaseName(), configEntry.getTableName()));
                configEntry.setEsCount(elasticsearchService.count(configEntry.getEsIndex(), configEntry.getEsType()));
                configEntry.setChangedCount(0L);
           // }
        });
        syncMappingConfigList.sort((config1, config2)->((Long)config2.getMysqlCount()).compareTo(config1.getMysqlCount()));
        long configId = 0L;
        for(SyncMappingConfig configEntry:syncMappingConfigList) {
            configEntry.setId(configId++);
        }
        logger.debug("结束刷新【{}】", System.currentTimeMillis());
        return syncMappingConfigList;
    }

    public void updateSyncMappingConfig(String dbName, String tableName, int changeCount) {
        logger.info(String.format("更新map=%s, database=%s,table=%s, count=%s完成", syncMappingConfigMap,dbName, tableName, changeCount));
        SyncMappingConfig configEntry = syncMappingConfigMap.get(dbName+"."+tableName);
        configEntry.setChangedCount(configEntry.getChangedCount() + changeCount);
    }

    public void setDbEsMapping(Map<String, String> dbEsMapping) {
        this.dbEsMapping = dbEsMapping;
    }

    public Map<String, String> getDbEsDataTypeMapping() {
        return dbEsDataTypeMapping;
    }

    public void setDbEsDataTypeMapping(Map<String, String> dbEsDataTypeMapping) {
        this.dbEsDataTypeMapping = dbEsDataTypeMapping;
    }

    public ElasticsearchService getElasticsearchService() {
        return elasticsearchService;
    }

    public void setElasticsearchService(ElasticsearchService elasticsearchService) {
        this.elasticsearchService = elasticsearchService;
    }
    public List<SyncMappingConfig> getSyncMappingConfigList() {
        return syncMappingConfigList;
    }

    public void setSyncMappingConfigList(List<SyncMappingConfig> syncMappingConfigList) {
        this.syncMappingConfigList = syncMappingConfigList;
    }

    public Map<String, SyncMappingConfig> getSyncMappingConfigMap() {
        return syncMappingConfigMap;
    }

    public void setSyncMappingConfigMap(Map<String, SyncMappingConfig> syncMappingConfigMap) {
        this.syncMappingConfigMap = syncMappingConfigMap;
    }

    private interface Converter {
        Object convert(String data);
    }
}
